/*
 *  pyzookeeper.c
 *  skywarp
 *
 *  Created by Aaron Cordova on 2/9/09.
 *  Copyright 2009 Aaron Cordova. All rights reserved.
 *
 *  Permission is hereby granted, free of charge, to any person obtaining a copy
 *  of this software and associated documentation files (the "Software"), to deal
 *  in the Software without restriction, including without limitation the rights
 *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 *  copies of the Software, and to permit persons to whom the Software is
 *  furnished to do so, subject to the following conditions:
 * 
 *  The above copyright notice and this permission notice shall be included in
 *  all copies or substantial portions of the Software.
 * 
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 *  THE SOFTWARE.
 */

#include "_pykeeper.h"
#include "zookeeper.jute.h"

#define _LL_CAST_ (long long)

static PyObject *_PyKeeperError;
static const char *clientIdFile = 0;

// convert error messages into slightly more useful messages
static char errorString[100];
static void Z_API_ErrorString(int err, char *string) {
    switch(err) {
        case ZNONODE:
            sprintf(string, "no node");
            break;
        case ZNOAUTH:
            sprintf(string, "no auth");
            break;
        case ZBADVERSION:
            sprintf(string, "bad version");
            break;
        case ZNOCHILDRENFOREPHEMERALS:
            sprintf(string, "no children for ephemerals");
            break;
        case ZNODEEXISTS:
            sprintf(string, "node exists");
            break;
        case ZNOTEMPTY:
            sprintf(string, "not empty");
            break;
        case ZSESSIONEXPIRED:
            sprintf(string, "session expired");
            break;
        case ZINVALIDCALLBACK:
            sprintf(string, "invalid callback");
            break;
        case ZINVALIDACL:
            sprintf(string, "invalid ACL");
            break;
        case ZAUTHFAILED:
            sprintf(string, "auth failed");
            break;
        case ZCLOSING:
            sprintf(string, "closing");
            break;
        case ZNOTHING:
            sprintf(string, "nothing");
            break;
    }
}

// generic watcher - calls the python watcher function
void watcher(zhandle_t *zh, int type, int state, const char *path, void* context) {
    
    // now just call the python callback method
    fprintf(stderr,"Watcher %d state = %d for %s\n", type, state, (path ? path: "null"));
    
    int pkType;

    // This is only generated by watches on non-existent nodes. These watches
    // are set using \ref zoo_exists.
    if(type == ZOO_CREATED_EVENT) {
        pkType = PK_CREATED_EVENT;
    }
    
    // This is only generated by watches on nodes. These watches
    // are set using \ref zoo_exists and \ref zoo_get.
    else if(type == ZOO_DELETED_EVENT) {
        pkType = PK_DELETED_EVENT;
    }
    
    // This is only generated by watches on nodes. These watches
    // are set using \ref zoo_exists and \ref zoo_get.
    
    else if(type == ZOO_CHANGED_EVENT) {
        pkType = PK_CHANGED_EVENT;
    }
    
    // This is only generated by watches on the child list of a node. These watches
    // are set using \ref zoo_get_children.
    else if(type == ZOO_CHILD_EVENT) {
        pkType = PK_CHILD_EVENT;
    }
    
     // This is generated when a client loses contact or reconnects with a server.
    else if(type == ZOO_SESSION_EVENT) {
        pkType = PK_SESSION_EVENT;
    
        if (state == ZOO_CONNECTED_STATE) {
        } 
        
        else if (state == ZOO_AUTH_FAILED_STATE) {
            fprintf(stderr, "Authentication failure. Shutting down...\n");
            zookeeper_close(zh);
            return;
        } 
        
        else if (state == ZOO_EXPIRED_SESSION_STATE) {
            fprintf(stderr, "Session expired. Shutting down...\n");
            zookeeper_close(zh);
            return;
        }
    }
    
    // This is generated when the server for some reason, probably a resource
    // constraint, will no longer watch a node for a client.
    else if(type == ZOO_NOTWATCHING_EVENT) {
        pkType = PK_NOTWATCHING_EVENT;
    }
    
    printf("got type: %d", pkType);
    
    if(context != NULL) {
        printf("ensuring state \n");
        PyGILState_STATE gstate;
        gstate = PyGILState_Ensure();
        
        printf("watcher: building arglist ...\n");
        PyObject *arglist = Py_BuildValue("(iis)", pkType, state, (path ? path: "no path"));
        
        printf("watcher: calling callback at %d...\n", context);
        PyObject *result = PyEval_CallObject((PyObject *)context, arglist);
        
        Py_DECREF(arglist);
        if(result)
            Py_DECREF(result);
        else 
            printf("error: result was null");
        
        printf("releasing state\n");
        PyGILState_Release(gstate);
    }
}

// this seems to be executed in the main thread
void my_strings_completion(int rc, const struct String_vector *stringv, const void *data) {

    int i;
    callback_info *in = (callback_info *)data;
    
    in->data = NULL;
    
    if(stringv) {        
        in->data = PyList_New(0);
        
        for(i=0; i < stringv->count; i++) {
            PyObject *string = PyString_FromString(stringv->data[i]);
            PyList_Append(in->data, string);
            Py_INCREF(string);
        }
        
    }
    
    in->complete = 1;
}

/*
void my_string_completion(int rc, const char *value, const void *data) {
    printf("in my string completion\n");
    callback_info *in = (callback_info *)data;
    
    in->data = NULL;
    
    if(value) {
        printf("sc: ensuring state \n");
//        PyGILState_STATE gstate;
//        gstate = PyGILState_Ensure();
        
        printf("sc: setting return value\n");    
//        in->data = PyString_FromString(value);
        
        printf("sc: releasing state \n");
//        PyGILState_Release(gstate);
    }
    
    in->complete = 1;
}*/

void my_stat_completion(int rc, const struct Stat *stat, const void *data) {
    
    callback_info *e_info = (callback_info *)data;
    
    if (!stat) {
        fprintf(stderr,"stat: null\n");
        Py_INCREF(Py_False);
        e_info->data = Py_False;
        e_info->complete = 1;
        return;
    }

    fprintf(stderr, "\t version=%x \t aversion=%x\n \t ephemeralOwner = %llx \n",
            (unsigned int)stat->version, 
            (unsigned int)stat->aversion, 
            _LL_CAST_ stat->ephemeralOwner);
    
    Py_INCREF(Py_True);
    e_info->data = Py_True;
    e_info->complete = 1;
}

void my_data_completion(int rc, const char *value, int value_len,
                        const struct Stat *stat, const void *data) {

    callback_info *d_info = (callback_info *)data;
    
    if (value) {
        //fprintf(stderr, " value_len = %d\n", value_len);
        //assert(write(2, value, value_len) == value_len);
        // should be a string
        PyObject *vString = PyString_FromString(value);
        d_info->data = vString;
    }
    else {
        Py_INCREF(Py_None);
        d_info->data = Py_None;
    }
    
    d_info->complete = 1;
}


/*
 * this one allows us to call a function defined in python as the watcher
 *
 */

static PyObject *_pykeeper_init(PyObject *self, PyObject *args) {
    const char *hostPort;
    PyObject *func;
    
    if(!PyArg_ParseTuple(args, "sO:func", &hostPort, &func))
        return NULL;
    
    if(!PyCallable_Check(func)) {
        PyErr_SetString(PyExc_TypeError, "parameter must be callable");
        return NULL;
    }
    
    Py_XINCREF(func);
      
    //printf("connecting to zookeeper \n");
    
    zhandle_t *zh = zookeeper_init("localhost:2181", watcher, 10000, 0, func, 0);
    
    if (!zh) {
        PyErr_SetString(_PyKeeperError, "could not connect\n");
        return NULL; 
    }
    
    const clientid_t *cid = zoo_client_id(zh);
    PyObject *handle = Py_BuildValue("ii", (int)zh, cid->client_id);
    return handle;
}

static PyObject *_pykeeper_close(PyObject *self, PyObject *args) {
    int handle;
    
    if (!PyArg_ParseTuple(args, "i", &handle))
        return NULL;
    
    int rc = zookeeper_close((zhandle_t *)handle);
    
    Py_RETURN_NONE;
}

static PyObject *_pykeeper_create(PyObject *self, PyObject *args) {
    const char *path;
    const char *value;
    int handle;
    int pflags;
    int valueLen;
    
    char realPath[100];
    
    if (!PyArg_ParseTuple(args, "issii", &handle, &path, &value, &valueLen, &pflags))
        return NULL;
    
    int flags = 0;
    if(pflags && PK_CREATE_SEQUENCE)
        flags |= ZOO_SEQUENCE;
    
    if(pflags && PK_CREATE_EPHEMERAL)
        flags |= ZOO_EPHEMERAL;
    
    //callback_info create_info;
    //create_info.complete = 0;
    
    int rc = zoo_create((zhandle_t *)handle, path, value, valueLen, &ZOO_OPEN_ACL_UNSAFE, flags, realPath, 100);
    if (rc) {
        Z_API_ErrorString(rc, errorString);
        PyErr_SetString(_PyKeeperError, errorString);
        return NULL; 
    }
    
    PyObject *rpString = PyString_FromString(realPath);
    return rpString;
    //while(create_info.complete == 0)
    //    usleep(200);
    
    //printf("string completion set complete\n");
    //if(create_info.data != NULL) {
    //    return create_info.data;
    //}
    
    //Py_RETURN_NONE;
}


static PyObject *_pykeeper_delete(PyObject *self, PyObject *args) {
    int handle;
    const char *path;
    int version;
    
    if (!PyArg_ParseTuple(args, "isi", &handle, &path, &version))
        return NULL;
    
    int rc = zoo_delete((zhandle_t *)handle, path, version);
    if (rc) { 
        Z_API_ErrorString(rc, errorString);
        PyErr_SetString(_PyKeeperError, errorString);
        return NULL; 
    }
    
    Py_RETURN_NONE;
}


static PyObject *_pykeeper_get_children(PyObject *self, PyObject *args) {
    int handle;
    const char *path;
    int watch;
    
    callback_info c_info;
    
    if (!PyArg_ParseTuple(args, "isi", &handle, &path, &watch))
        return NULL;
    
    struct String_vector *stringv;
    c_info.complete = 0;
    
    int rc = zoo_aget_children((zhandle_t*)handle, path, watch, my_strings_completion, &c_info);

    if (rc) {
        Z_API_ErrorString(rc, errorString);
        PyErr_SetString(_PyKeeperError, errorString);
        return NULL; 
    }
    
    // this is crappy but the only way this c client seems to work
    // synchronous call to ls doesn't seem to work (stringv empty)
    while(c_info.complete == 0)
        usleep(200);
    
    if(c_info.data != NULL) {
        return c_info.data;
    }
    
    Py_RETURN_NONE;
}

static PyObject *_pykeeper_exists(PyObject *self, PyObject *args) {
    int handle;
    const char *path;
    int watch;
    
    if (!PyArg_ParseTuple(args, "isi", &handle, &path, &watch))
        return NULL;
    
    callback_info e_info;
    e_info.complete = 0;
    
    struct Stat stat;
    
    int rc = zoo_aexists((zhandle_t *)handle, path, watch, my_stat_completion, &e_info);
    //int rc = zoo_exists((zhandle_t *)handle, path, watch, &stat);
    if (rc) {
        Z_API_ErrorString(rc, errorString);
        PyErr_SetString(_PyKeeperError, errorString);
        return NULL; 
    }
    
    while(e_info.complete == 0) 
        usleep(200);
    
    return e_info.data;
}

static PyObject *_pykeeper_get_data(PyObject *self, PyObject *args) {
    int handle;
    const char *path;
    int watch;
    
    if (!PyArg_ParseTuple(args, "isi", &handle, &path, &watch))
        return NULL;
    
    callback_info d_info;
    d_info.complete = 0;
    
    int rc = zoo_aget((zhandle_t *)handle, path, watch, my_data_completion, &d_info);
    if (rc) {
        Z_API_ErrorString(rc, errorString);
        PyErr_SetString(_PyKeeperError, errorString);
        return NULL; 
    }
    
    while(d_info.complete == 0) 
        usleep(200);
    
    return d_info.data;
}

static PyObject* _pykeeper_set_log_level(PyObject *self, PyObject *args) {
    int level;
    
    if (!PyArg_ParseTuple(args, "i", &level))
        return NULL;
    
    zoo_set_debug_level(level);
    
    Py_RETURN_NONE;
}

static PyMethodDef PyKeeperMethods[] = {
    {"init",  _pykeeper_init, METH_VARARGS, "Initiate Zookeeper connection."},
    {"close", _pykeeper_close, METH_VARARGS, "Close a zookeeper session"},
    {"create", _pykeeper_create, METH_VARARGS, "Create a node"},
    {"delete", _pykeeper_delete, METH_VARARGS, "Delete a node"},
    {"getChildren",  _pykeeper_get_children, METH_VARARGS, "List the childred of a node"},
    {"exists", _pykeeper_exists, METH_VARARGS, "Return whether a node exists"},
    {"getData",  _pykeeper_get_data, METH_VARARGS, "Return the data associated with a node"},
    {"setLogLevel", _pykeeper_set_log_level, METH_VARARGS, "Set level of logging detail"},  
    {NULL, NULL, 0, NULL}        /* Sentinel */
};


PyMODINIT_FUNC init_pykeeper(void) {
    PyObject *m;

    m = Py_InitModule("_pykeeper", PyKeeperMethods);
    if (m == NULL)
        return;
    
    _PyKeeperError = PyErr_NewException("_pykeeper.error", NULL, NULL);
    Py_INCREF(_PyKeeperError);
    PyModule_AddObject(m, "error", _PyKeeperError);

    PyEval_InitThreads();
    
    //zoo_set_debug_level(ZOO_LOG_LEVEL_DEBUG);
    zoo_deterministic_conn_order(1); // enable deterministic order
}


